#include "netComponents.h"
#include "tools/cJSON.h"
#include <QtDebug>
#include <QDateTime>

#define  NODESERVER_ACCEPT_BUF_LEN    1024

nodeServer::nodeServer()
        : m_bExit(false)
        , m_bInitStatus(false)
        , m_sockServer(INVALID_SOCKET)
{
    m_strNodeKey.clear();
    m_mapProcClients.clear();
    m_listNode.clear();
    m_listDispatch.clear();
    m_listMsgInfos.clear();
}

nodeServer::~nodeServer()
{
    release();
}

// 初始化服务端，配置口令、端口和侦听个数
bool nodeServer::init(std::string strNodeKey, unsigned int uiPort, unsigned int uiListenSize)
{
    m_bInitStatus = false;
    do
    {
        if(strNodeKey.empty() || 0 == uiPort || 0 == uiListenSize)
        {
            break;
        }

        m_strNodeKey.assign(strNodeKey);
        m_sockServer = INVALID_SOCKET;
        m_sockServer = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

        if(INVALID_SOCKET == m_sockServer)
        {
            qDebug() <<"[nodeServer::init] socket create faild";
            break;
        }

        struct sockaddr_in stServerAddr;
        stServerAddr.sin_family = AF_INET;
        stServerAddr.sin_port = htons(uiPort);
        stServerAddr.sin_addr.S_un.S_addr = INADDR_ANY;
        if(SOCKET_ERROR == bind(m_sockServer, (LPSOCKADDR)&stServerAddr, sizeof (stServerAddr)))
        {
            qDebug() <<"[nodeServer::init] socket bind faild";
            break;
        }

        if(SOCKET_ERROR == listen(m_sockServer, uiListenSize))
        {
            qDebug() <<"[nodeServer::init] socket listen faild";
            break;
        }

        qDebug() << "[nodeServer::init] socket create/bind/listen success!!";

        // 创建侦听线程
        try
        {
            m_thListenProc = std::thread(&nodeServer::threadListenProc, this);
            m_thNodeMgr = std::thread(&nodeServer::threadNodeMgr, this);
            m_thDispatch = std::thread(&nodeServer::threadDispatch, this);
        }
        catch (...)
        {
            qDebug() <<"[nodeServer::init] socket thread create faild";
            break;
        }

        m_bInitStatus = true;
    } while (0);
    
    if(false == m_bInitStatus && INVALID_SOCKET != m_sockServer)
    {
        closesocket(m_sockServer);
        m_sockServer = INVALID_SOCKET;
    }

    return m_bInitStatus;
}

// 获取服务端状态
bool nodeServer::getStatus()
{
    return m_bInitStatus;
}

// 释放资源
void nodeServer::release()
{
    m_bExit = true;

    // 通知消息数据调度线程退出
    {
        std::unique_lock<std::mutex> lck (m_mtxDispatch);
        m_conDispatch.notify_all();
    }

    {
        std::unique_lock<std::mutex> lck (m_mtxNodes);
        m_conNodes.notify_all();
    }

    if(m_thListenProc.joinable())
    {
        m_thListenProc.join();
    }

    if(m_thDispatch.joinable())
    {
        m_thDispatch.join();
    }

    if(m_thNodeMgr.joinable())
    {
        m_thNodeMgr.join();
    }
}

// 连接侦听线程
void nodeServer::threadListenProc()
{
    qDebug() << "[nodeServer::threadListenProc] working ...";

    /*
     *  客户端连接规则：创建两个连接，第一个为登录，第二个为心跳，均有客户端负责创建
     *  侦听线程在接收到两者的连接之后，创建统一符合节点对应服务
     *  对于不符合的，则不创建服务节点
    */
    while (!m_bExit)
    {
        CLIENTINFO stClientInfo;
        stClientInfo.nSocket = INVALID_SOCKET;
        struct timeval tv;
        tv.tv_sec = 1;  // 改为一秒，方便退出主进程
        tv.tv_usec = 0;
        fd_set  stFdSet;
        FD_ZERO(&stFdSet);
        FD_SET(m_sockServer, &stFdSet);
        if(select(m_sockServer, &stFdSet, nullptr, nullptr, &tv) <= 0)
        {
            //qDebug() << "[nodeServer::threadListenProc] select error, conintue, time: " << tv.tv_sec;
            continue;
        }

        struct sockaddr_in stAddClient;
        socklen_t addlen = sizeof (struct sockaddr_in);
        stClientInfo.nSocket = accept(m_sockServer, (struct sockaddr *)&stAddClient, &addlen);

        qDebug() << "[nodeServer::threadListenProc] accept, nSocket: " << stClientInfo.nSocket;

        if(INVALID_SOCKET == stClientInfo.nSocket)
        {
            qDebug() << "[nodeServer::threadListenProc] accept error, conintue";
            continue;
        }

        stClientInfo.strAddr = inet_ntoa(stAddClient.sin_addr); // 获取客户端IP地址

        // 接收报文数据并判断类型，确认是否连接
        char revData[NODESERVER_ACCEPT_BUF_LEN];
        int iRet = recv(stClientInfo.nSocket, revData, NODESERVER_ACCEPT_BUF_LEN, 0);
        //qDebug() << QString("[nodeServer::threadListenProc] %1 recv len %2 ").arg(QString::fromStdString(stClientInfo.strAddr), QString::number(iRet));
        if(iRet <= 0)
        {
            qDebug() << "[nodeServer::threadListenProc] recv key error, conintue, iRet: " << iRet;
            continue;
        }

        //qDebug() << "[nodeServer::threadListenProc] recv done!, ready to send ack, strAddr: " << QString::fromStdString(stClientInfo.strAddr);

        getConnectType(revData, stClientInfo);
        if(stClientInfo.strCheckKey == m_strNodeKey)
        {
            // 校验通过，执行节点添加
            std::string strAck;
            if(false == getAckInfo(true, strAck))
            {
                qDebug() << QString("[nodeServer::threadListenProc] %1 getAckInfo faild, continue").arg(QString::fromStdString(stClientInfo.strAddr));
                continue;
            }
            send(stClientInfo.nSocket, strAck.c_str(), strAck.length(), 0); // 应答连接成功

            addNodeProc(stClientInfo);
        }
        else
        {
            // 校验不通过
            std::string strAck;
            if(false == getAckInfo(false, strAck))
            {
                qDebug() << "[nodeServer::threadListenProc] getAckInfo faild, continue, strAddr: " << QString::fromStdString(stClientInfo.strAddr);
                continue;
            }

            send(stClientInfo.nSocket, strAck.c_str(), strAck.length(), 0); // 应答连接成功
        }
    }
}

// 节点管理线程
void nodeServer::threadNodeMgr()
{
    while (!m_bExit)
    {
        std::list<MSG_INFO> listNodes;
        {
            std::unique_lock<std::mutex> lck (m_mtxNodes);
            m_conNodes.wait_for(lck, std::chrono::seconds(NODEPROC_WAITTIME_S), [&](){return (!m_listNode.empty() || m_bExit);});
            if(!m_listNode.empty())
            {
                m_listNode.swap(listNodes);
            }
        }

        if(m_bExit)
        {
            break;
        }

        // 针对掉线节点，先删除该节点
        {
            std::unique_lock<std::mutex> lck (m_mtxNodes);
            for(auto it : listNodes)
            {
                // 节点掉线或是主动退出，删除该节点
                if(MSG_INFO_TYPE_STATUS_OFF == it.nType || MSG_INFO_TYPE_STATUS_EXIT == it.nType )
                {
                    auto itNode = m_mapProcClients.find(it.strID);
                    if(m_mapProcClients.end() != itNode)
                    {
                        delete itNode->second;
                        m_mapProcClients.erase(itNode);
                    }
                }
            }
        }

        // 将所有信息打包成报文，发送到各个节点， 各个节点仅解析非本节点的状态并上报展示
        NODE_MSG_INFO  stDispatch;
        stDispatch.strNode.assign(gstrComNodeStatus);
        if(false == createSendMsg(listNodes, stDispatch.strMsg))
        {
            qDebug() << "[nodeServer::threadNodeMgr] createSendMsg faild, continue";
            continue;
        }

        // 将状态报文发送到各个节点上去
        if(!stDispatch.strMsg.empty())
        {
            std::unique_lock<std::mutex> lck (m_mtxDispatch);
            m_listDispatch.push_back(stDispatch);
            m_conDispatch.notify_all();
        }
    }
}

// 数据调度线程
void nodeServer::threadDispatch()
{
    while (!m_bExit)
    {
        std::list<NODE_MSG_INFO>  listDispatch;
        {
            std::unique_lock<std::mutex> lck (m_mtxDispatch);
            m_conDispatch.wait_for(lck, std::chrono::seconds(NODEPROC_WAITTIME_S), [&](){return (!m_listDispatch.empty() || m_bExit);});
            if(!m_listDispatch.empty())
            {
                m_listDispatch.swap(listDispatch);
            }
        }

        if(m_bExit)
        {
            break;
        }

        if(listDispatch.empty())
        {
            continue;
        }

        // 需要回调上层展示的数据，进行解析后通知
        if(true == m_bCallBackStatus)
        {
            // 逐个解析数据后通知上层获取数据
            for(auto it : listDispatch)
            {
                std::list<MSG_INFO> listMsg;
                if(false == parseRecvMsg(it.strMsg, listMsg))
                {
                    continue;
                }

                if(!listMsg.empty())
                {
                    std::unique_lock<std::mutex> lck (m_mtxCallBackMsg);
                    m_listMsgInfos.insert(m_listMsgInfos.end(), listMsg.begin(), listMsg.end());
                    m_conCallBack.notify_all();
                }
            }
        }

        // 分发数据，将消息通知到除自身之外的其他处理端
        for(auto it : listDispatch)
        {
            std::unique_lock<std::mutex> lck (m_mtxClients);
            for(auto itNode : m_mapProcClients)
            {
                // 非本节点或状态报文，则发送出去
                if(itNode.first != it.strNode || gstrComNodeStatus == it.strNode)
                {
                    itNode.second->insertMsg(it.strMsg);
                }
            }
        }
    }
}

// 添加节点
void nodeServer::addNodeProc(const CLIENTINFO &stClientInfo)
{
    qDebug() << "[nodeServer::addNodeProc] ready to add NodeProc, strAddr: " << QString::fromStdString(stClientInfo.strAddr);
    std::string strNodeKey = stClientInfo.strAddr + "_" + stClientInfo.strID;   // 节点ID以IP地址+ID形式保存
    bool bAddNewNode = false;
    {
        std::unique_lock<std::mutex> lck (m_mtxClients);
        auto itNode = m_mapProcClients.find(strNodeKey);
        if(m_mapProcClients.end() == itNode)
        {
            nodeProc *pNode = new nodeProc(this);
            if(nullptr != pNode)
            {
                m_mapProcClients.insert(std::make_pair(strNodeKey, pNode));
                pNode->initSocket(stClientInfo.nType, stClientInfo.nSocket);
            }
            else
            {
                qDebug() << "[nodeServer::addNodeProc] NodeProc new faild: " << QString::fromStdString(strNodeKey);
            }
        }
        else
        {
            nodeProc *pNode = itNode->second;
            if(nullptr != pNode)
            {
                // 插入套接字成功，则初始化处理节点
                if(true == pNode->initSocket(stClientInfo.nType, stClientInfo.nSocket))
                {
                    if(true == pNode->init(strNodeKey))
                    {
                        // 通知上层有新节点加入
                        qDebug() << "[nodeServer::addNodeProc] ready to notify information";
                        bAddNewNode = true;
                    }
                    else
                    {
                        qDebug() << "[nodeServer::addNodeProc] NodeProc init faild";
                    }
                }
            }
            else
            {
                qDebug() << "[nodeServer::addNodeProc] NodeProc found faild :" << QString::fromStdString(strNodeKey);
            }
        }
    }

    if(true == bAddNewNode)
    {
        MSG_INFO stMsg;
        stMsg.nType = MSG_INFO_TYPE_STATUS_NEW;
        stMsg.strTime = QDateTime::currentDateTime().toString("yyyy-MM-dd hh:mm:ss").toStdString();
        stMsg.strID.assign(strNodeKey);

        {
            std::unique_lock<std::mutex> lck (m_mtxNodes);
            m_listNode.emplace_back(stMsg);
            m_conNodes.notify_all();
        }
    }
}

// 从连接登录报文中解析套接字类型
void nodeServer::getConnectType(std::string strMsg, CLIENTINFO &stClientInfo)
{
    stClientInfo.nType = CONNECT_TYPE_UNKNOW;
    stClientInfo.strID.clear();
    cJSON *pRoot = nullptr;

    do
    {
        if(strMsg.empty())
        {
            break;
        }

        pRoot = cJSON_Parse(strMsg.c_str());
        if(nullptr == pRoot)
        {
            break;
        }

        cJSON *pnetType = cJSON_GetObjectItem(pRoot, NET_DES_NETTYPE);
        cJSON *pnodeID = cJSON_GetObjectItem(pRoot, NET_DES_NODEID);
        cJSON *pnodeKey = cJSON_GetObjectItem(pRoot, NET_DES_CHECKKEY);
        if( nullptr == pnetType || nullptr == pnodeID || nullptr == pnodeKey)
        {
            break;
        }

        stClientInfo.nType = (CONNECT_TYPE)pnetType->valueint;
        stClientInfo.strID.assign(pnodeID->valuestring);
        stClientInfo.strCheckKey.assign(pnodeKey->valuestring);

    }while(0);

    if(nullptr != pRoot)
    {
        cJSON_Delete(pRoot);
        pRoot = nullptr;
    }
}

// 生成应答报文
bool nodeServer::getAckInfo(const bool bStatus, std::string &strAck)
{
    bool bRet = false;
    cJSON *pRoot = nullptr;
    char* pRet = nullptr;
    do
    {
        strAck.clear();
        pRoot = cJSON_CreateObject();
        if(nullptr == pRoot)
        {
            break;
        }

        //cJSON_AddStringToObject(pRoot, NET_DES_NODEID, m_strNodeID.c_str());    // 服务节点ID
        cJSON_AddBoolToObject(pRoot, NET_DES_LINK, bStatus);                    // 连接状态
        pRet = cJSON_PrintUnformatted(pRoot);
        if(nullptr != pRet)
        {
            strAck.assign(pRet);
        }
        else
        {
            break;
        }

        bRet = true;
    }while(0);

    if(nullptr != pRoot)
    {
        cJSON_Delete(pRoot);
        pRoot = nullptr;
    }

    if(nullptr != pRet)
    {
        free(pRet);
        pRet = nullptr;
    }

    return bRet;
}

// 子处理节点状态回调
void nodeServer::funcDataCallbackStatus(const MSG_INFO_TYPE nType, const std::string &strNodeID)
{
    if(strNodeID.empty() || !(MSG_INFO_TYPE_STATUS_OFF == nType || MSG_INFO_TYPE_STATUS_EXIT == nType ))
    {
        return;
    }

    MSG_INFO stMsg;
    stMsg.nType = nType;
    stMsg.strTime = QDateTime::currentDateTime().toString("yyyy-MM-dd hh:mm:ss").toStdString();
    stMsg.strID.assign(strNodeID);

    {
        std::unique_lock<std::mutex> lck (m_mtxNodes);
        m_listNode.emplace_back(stMsg);
        m_conNodes.notify_all();
    }
}

// 数据回调
void nodeServer::funcDataCallbackMsg(const std::string &strNodeID, std::string &strMsgInfo)
{
    //qDebug() << "[nodeServer::funcDataCallbackMsg] NodeID:" << QString::fromStdString(strNodeID);
    if(strNodeID.empty() || strMsgInfo.empty())
    {
        return;
    }

    NODE_MSG_INFO  stDispatch;
    stDispatch.strNode.assign(strNodeID);
    stDispatch.strMsg.assign(strMsgInfo);
    {
        std::unique_lock<std::mutex> lck (m_mtxDispatch);
        m_listDispatch.push_back(stDispatch);
        m_conDispatch.notify_all();
    }
}